package com.atguigu.flink.charkoer02;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.client.program.StreamContextEnvironment;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class FlinkWordCount {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();

        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamContextEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);

        DataStreamSource<String> source = env.socketTextStream("hadoop162", 9999);

        source.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                String[] line = value.split(" ");
                for (String word : line) {
                    out.collect(word);
                }
            }
        }).map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return Tuple2.of(value, 1);
            }
        }).keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> value) throws Exception {
                return value.f0;
            }
        }).sum(1).print();

        env.execute();

    }
}
/*
如何修改算子的并行度
1. 在配置中文件中指定
    parallelism.default: 1

2. 在提交job的时候, 通过 -p 参数指定
    -p 2

3. 在代码中, 通过env来指定job的所有算子的并行度
    env.setParallelism(1);

4. 在代码中,单独给算子设置并行度
   .setParallelism(1);

------
.startNewChain();
    当前算子不和前面的算子优化在一起
.disableChaining()
    这个算子单独
 env.disableOperatorChaining();
    全局禁用

.
 */